# -*- coding: utf-8 -*-
# __author__ = 'Yuanjiang Huang'
# yuanjiang.huang@socialcredits.cn

import os
import sys

# Path for spark source folder
os.environ['SPARK_HOME']="/home/openkai/social-credits/spark/spark-1.5.1/"

# Append pyspark  to Python Path
sys.path.append("/home/openkai/social-credits/spark/spark-1.5.1/python/")

try:
	from pyspark import SparkContext
	from pyspark import SparkConf
	print ("Successfully imported Spark Modules")
except ImportError as e:
    print ("Can not import Spark Modules", e)
    sys.exit(1)

CURRENT_PATH = os.path.dirname(__file__)
if CURRENT_PATH:
    CURRENT_PATH = CURRENT_PATH + '/'


def containLine(s):
	return 'line' in s

if __name__ == "__main__":
	# if master_url is local, it tells Spark how to connect to a cluster.
	master_url = 'spark://sc-data-server-1:7077' #one worker thread
	# master_url = 'local'
	conf = SparkConf().setMaster(master_url).setAppName('Demo')
	sc = SparkContext(conf=conf)
	data = 'file:///'+'home/openkai/social-credits/spark-tutorial'+'/data/README.txt'

	# create a RDD
	lines = sc.textFile(data)
	print '*'*200
	print 'lines type:',type(lines)
	print '*'*200
	#filter transform RDD
	newLines = lines.filter(lambda line: "socialcredits" in line)
	print 'new lines type: ',type(newLines)
	print '*'*200

	#action on RDD, get the first line and then return to the driver program
	first_line = newLines.first()
	print '*'*200
	print 'first line: ', type(first_line)
	print first_line
	print '*'*200
	#
	# # action on RDD, get the number of lines:
	# num = newLines.count()
	# print 'the number of lines is: ',num
	#
	# #filter transformation on 'lines'
	# newlines2 = lines.filter(containLine)
	#
	# #action on RDD, get all items
	# results = newlines2.collect()
	# print results
	#
	# #transform on RDD
	# pairs = lines.map(lambda x: (x.split(" ")[1], x.split(" ")[0]))
	# #action on RDD
	# results = pairs.collect()
	# print dict(results)
	#
	# # create a RDD
	# data = sc.parallelize([1,2,3,3,4])
	#
	# #action on RDD
	# count_by_value = data.countByValue()
	# print '*'*200
	# print dict(count_by_value )
	# print '*'*200
